package connection

import (
	"broker/interface/imodel"
	"broker/interface/iserver"
	"broker/struct/datapack"
	"broker/struct/model"
	"broker/struct/request"
	"broker/utils"
	"broker/utils/logger"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
)

/**
连接模块
*/
type Connection struct {
	//当前Conn隶属Server
	TcpServer iserver.IServer

	//当前连接的socket TCP套接字
	Conn *net.TCPConn

	//当前连接的ID
	ConnID uint32

	//当前连接的状态
	isClosed bool

	//告知当前连接已经退出/停止的channel (Reader 告知 Writer)
	ExitChan chan bool

	//无缓冲管道，用于读、写Goroutine之间消息通信
	msgChan chan []byte

	//该链接处理的router
	MsgHandle iserver.IMsgHandle

	//连接属性
	property map[string]interface{}
	//连接属性的读写锁
	propertyLock sync.RWMutex
}

func NewConnection(server iserver.IServer, conn *net.TCPConn, connID uint32, msgHandle iserver.IMsgHandle) *Connection {
	c := &Connection{
		TcpServer: server,
		Conn:      conn,
		ConnID:    connID,
		isClosed:  false,
		msgChan:   make(chan []byte),
		ExitChan:  make(chan bool, 1),
		MsgHandle: msgHandle,
		property:  map[string]interface{}{},
	}
	//将conn加入连接管理器
	c.TcpServer.GetConnManager().AddConnection(c)
	return c
}

// StartReader 连接的读业务方法
func (c *Connection) StartReader() {
	logger.Info.Println("[Connect] Reader Goroutine is running...")
	defer logger.Info.Println("[Connect] ConnId = ", c.ConnID, "Reader is exit,remote addr is ", c.RemoteAddr().String())
	defer c.Stop()

	for {
		//创建一个拆包解包的对象
		dp := datapack.NewBaseDataPack()

		//读取消息ID
		messageHeader, err := dp.GetMessageHeader(c)
		if c.isClosed {
			break
		}
		if err != nil {
			if err == io.EOF {
				logger.Info.Println("[Connect] disconnect")
			} else {
				logger.Error.Println("[Connect] read header fail", err)
			}
			break
		}
		//创建一个消息数据
		msg := model.NewEmptyBaseMessage()
		msg.SetMsgId(messageHeader >> 4)
		msg.SetReserved((messageHeader << 4) >> 4)
		//读取数据长度
		messageLen, err := dp.GetDataLen(c)
		if err != nil {
			logger.Error.Println("[Connect] read msg len fail", err)
			break
		}
		msg.SetMsgLen(messageLen)

		//得到dataLen，再读取Data，放在msg.Data里
		var data []byte
		if msg.GetMsgLen() > 0 {
			data = make([]byte, msg.GetMsgLen())

			if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {
				logger.Error.Println("[Connect] read msg data fail", err)
				break
			}

		}
		msg.SetData(data)

		//得到当前conn数据的Request请求数据
		req := request.Request{
			Conn: c,
			Msg:  *msg,
		}

		if utils.GetConfig().WorkerPool {
			//将request添加到消息队列工作池
			c.MsgHandle.SendMsgToTaskQueue(&req)
		} else {
			go c.MsgHandle.DoMsgHandle(&req)
		}
	}

}

/*
	写消息Goroutine，专用于发送给客户端消息的模块
*/
func (c *Connection) StartWriter() {
	logger.Info.Println("[Connect] Writer Goroutine is running")
	defer logger.Info.Println("[Connect] ConnId = ", c.ConnID, "Writer is exit,remote addr is ", c.RemoteAddr().String())

	//循环阻塞等待channel消息，进行写给客户端
	for {

		select {
		case data := <-c.msgChan:
			if _, err := c.Conn.Write(data); err != nil {
				logger.Error.Println("[Connect] send data error", err)
				return
			}
		case <-c.ExitChan:
			//代表Reader已经退出，Writer也要退出
			return
		}
	}
}

// Start 启动连接 让当前连接准备开始工作
func (c *Connection) Start() {
	logger.Info.Println("[Connect] Conn Start... ConnId = ", c.ConnID)

	//启动从当前连接读数据业务
	go c.StartReader()
	//启动从当前连接写数据的业务
	go c.StartWriter()

}

// Stop 停止连接 结束当前连接的工作
func (c *Connection) Stop() {
	logger.Info.Println("[Connect] Conn Stop... ConnID = ", c.ConnID)
	//如果当前连接已经关闭
	if c.isClosed {
		return
	}
	c.isClosed = true

	//关闭socket连接
	err := c.Conn.Close()
	if err != nil {
		if err != nil {
			logger.Error.Println(fmt.Sprintf("连接 %d 关闭失败",c.ConnID))
		}

	}

	//告知Writer关闭
	c.ExitChan <- true

	//调用连接从连接管理器中移除Hook
	c.TcpServer.GetConnManager().RemoveConnection(c)
	err = c.TcpServer.GetTopicManager().RemoveSubscribe(model.NewSubscriber(c.ConnID, nil))
	if err != nil {
		logger.Error.Println(fmt.Sprintf("连接 %d 清除订阅失败",c.ConnID))
	}
	//回收资源
	close(c.ExitChan)
	close(c.msgChan)

}

// GetTCPConnection 获取当前连接的绑定socket conn
func (c *Connection) GetTCPConnection() *net.TCPConn {
	return c.Conn
}

// GetConnID 获取当前连接模块的连接ID
func (c *Connection) GetConnID() uint32 {
	return c.ConnID
}

// GetTcpServer 获取当前连接模块的连接ID
func (c *Connection) GetTcpServer() iserver.IServer {
	return c.TcpServer
}

// RemoteAddr 获取远程客户端的 TCP状态 IP port
func (c *Connection) RemoteAddr() net.Addr {
	return c.Conn.RemoteAddr()
}

// Send 直接发送数据，将数据发送给远程的客户端
func (c *Connection) Send(data []byte) error {
	c.msgChan <- data
	return nil
}

// SendMsg 封包发送数据
func (c *Connection) SendMsg(msg imodel.IMessage) error {
	if c.isClosed {
		return errors.New("[Connect] Connection closed when send msg")
	}
	//将data封包，msgDataLen,MsgId,Data

	dp := datapack.NewBaseDataPack()
	binaryMsg, err := dp.Pack(msg)
	if err != nil {
		logger.Error.Println("[Connect] pack err msg id =", msg.GetMsgId())
		return errors.New("pack msg error")
	}
	//将数据发送给客户端
	c.msgChan <- binaryMsg

	return nil
}

// SetProperty /*连接属性*/
//设置连接属性
func (c *Connection) SetProperty(key string, value interface{}) {
	c.propertyLock.Lock()
	defer c.propertyLock.Unlock()
	c.property[key] = value
}

// GetProperty 获取连接属性
func (c *Connection) GetProperty(key string) (interface{}, error) {
	c.propertyLock.RLock()
	defer c.propertyLock.RUnlock()
	if value, ok := c.property[key]; ok {
		return value, nil
	}
	return nil, errors.New("property not found")
}

// RemoveProperty 移除连接属性
func (c *Connection) RemoveProperty(key string) {
	c.propertyLock.Lock()
	defer c.propertyLock.Unlock()
	if _, ok := c.property[key]; ok {
		delete(c.property, key)
	}
}
